package tbdp.tas

import java.util.concurrent.TimeUnit

import org.apache.hadoop.hbase.client.{Put, HTableInterface, Result, Scan}
import org.apache.hadoop.hbase.filter.{PageFilter, FilterList}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import tbdp.tas.model.{MinsLine, StockBlock}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

/**
  * Created by xbkaishui on 16/9/5.
  * 处理板块 分时的处理逻辑
  */
class BKMinslineHandler(topic: String, brokers: String, interval: String, dt: String, bkInfos: ArrayBuffer[StockBlock] , oldVolRateTrainMap: mutable.HashMap[String, mutable.Map[String, ArrayBuffer[Double]]]) {
  /**
    * 从hbase表里面获取分时数据
    *
    * @return
    */
  def loadMinsDataFromHbase(): ArrayBuffer[MinsLine] = {
    val data: ArrayBuffer[MinsLine] = new ArrayBuffer[MinsLine]()
    val table = Utils.initHbaseTable(Constants.hbase_minsline)
    val scan = new Scan()
    val filterList = new FilterList()
    val pageFilter = new PageFilter(100000)
    filterList.addFilter(pageFilter)
    scan.setStartRow(Bytes.toBytes("0_"))
    scan.setFilter(filterList)

    val retryCnt = 3
    for (i <- 1 to retryCnt) {
//      println(i)
      try {
        val start = System.currentTimeMillis
        val resultScanner = table.getScanner(scan).iterator()
        val end = System.currentTimeMillis
//        println("scan table cost " + (end - start) + " ms")
        while (resultScanner.hasNext) {
          val r = resultScanner.next()
          try {
            val record: MinsLine = Utils.convert(r, table)
            //只取当天的股票
            if (record.tDate.equalsIgnoreCase(dt) && !Utils.isBk(record.marketType)) {
              data.append(record)
            }
          } catch {
            case e: Exception => {
              val key = Bytes.toString(r.getRow())
              println("error convert " + key)
            }
          }
        }
        return data
      } catch {
        case e: Exception => {
          e.printStackTrace()
        }
      }
    }
//    println("minline hbase data \n" + data)
    table.close()
    data
  }

  /**
    * 计算板块分时指标 板块分成综合板块和敏感板块
    * 敏感板块的处理逻辑是取 板块对应的股票的 按指标正序排序的前15个股票列表
    *
    * @return
    */
  def calcMinsLine(): (ArrayBuffer[MinsLine], Boolean) = {
    val start = System.currentTimeMillis()
    val sts: ArrayBuffer[MinsLine] = loadMinsDataFromHbase()
    val end = System.currentTimeMillis()
    println("load minsline streaming cost: "+(end-start))
    val stockMap = sts.map { t => (t.marketType + "_" + t.sCode, t) }.toMap

    val bkList = new ArrayBuffer[MinsLine]()

    val bkList3 = new ArrayBuffer[MinsLine]()
    val bkList2 = new ArrayBuffer[MinsLine]()
    bkInfos.foreach {
      bk => {
        val bkStock = MinsLine.fromBk(bk)
        bkStock.tDate = dt
        //sensitive bk MARKET TYPE IS 10
        val sensitiveBk = MinsLine.fromBk(bk)
        sensitiveBk.marketType = "10"
        sensitiveBk.codeType = "10"
        sensitiveBk.tDate = dt
        val stocks = bk.stocks
        val stockList = new ArrayBuffer[MinsLine]()
        //取出来当前板块对应的 股票列表
        stocks.foreach(id => {
          stockMap.get(id) match {
            case Some(st) => stockList.append(st)
            case _ =>
          }
        })

        if (stockList.nonEmpty) {
          calcBKStock(bkStock, stockList)
          calcSensitiveBk(sensitiveBk, stockList)
          bkList2.append(bkStock)
          bkList3.append(sensitiveBk)
        } else {
          println("empty stock list " + bkStock)
        }
      }
    }

//    println("calc agency1BSVolRateTrain")

    bkList.appendAll(bkList2.sortBy(-_.agency1BSVolRate).zipWithIndex.map(st => {
      st._1.agency1BSVolRateTrain = st._2.toInt+1
      st._1
    }))

    bkList.appendAll( bkList3.sortBy(-_.agency1BSVolRate).zipWithIndex.map(st => {
      st._1.agency1BSVolRateTrain = st._2.toInt+1
      st._1
    }))

    val stopFlag = checkNeedStop(sts)
    //更新状态
    if (stopFlag) {
      bkList.foreach(st => {
        st.tflag == "1"
      })
    }
//    println("bklist"+bkList)
//    println("stopFlag"+stopFlag)
    (bkList, stopFlag)
  }

  /**
    * 计算综合板块
    *
    * @param bkStock
    * @param stockList
    */
  def calcBKStock(bkStock: MinsLine, stockList: ArrayBuffer[MinsLine]): Unit = {
    bkStock.TTime = stockList(0).TTime.substring(0, 4)
    //计算分时逻辑
    bkStock.agencyBVol = stockList.map(_.agencyBVol).sum
    bkStock.agencyBAmt = stockList.map(_.agencyBAmt).sum
    bkStock.allBVol = stockList.map(_.allBVol).sum
    bkStock.allBAmt = stockList.map(_.allBAmt).sum
    bkStock.agencySVol = stockList.map(_.agencySVol).sum
    bkStock.agencySAmt = stockList.map(_.agencySAmt).sum
    bkStock.allSVol = stockList.map(_.allSVol).sum
    bkStock.allSAmt = stockList.map(_.allSAmt).sum

    bkStock.agencyBSco = stockList.map(_.agencyBSco).sum / stockList.size
    bkStock.agencySSco = stockList.map(_.agencySSco).sum / stockList.size
    bkStock.priTeamBSco = stockList.map(_.priTeamBSco).sum / stockList.size
    bkStock.priTeamSSco = stockList.map(_.priTeamSSco).sum / stockList.size
    bkStock.publicBSco = stockList.map(_.publicBSco).sum / stockList.size
    bkStock.publicSSco = stockList.map(_.publicSSco).sum / stockList.size
    bkStock.allSco = stockList.map(_.allSco).sum / stockList.size
    bkStock.avg5AllSco = stockList.map(_.avg5AllSco).sum / stockList.size

    bkStock.agencyBSAmt1D = stockList.map(_.agencyBSAmt1D).sum
    bkStock.agencyBSAmt3D = stockList.map(_.agencyBSAmt3D).sum
    bkStock.agencyBSAmt5D = stockList.map(_.agencyBSAmt5D).sum
    bkStock.agencyBSAmt7D = stockList.map(_.agencyBSAmt7D).sum

    //锁定率
    bkStock.agencyBVolRate = stockList.map(_.agencyBVolRate).sum / stockList.size
    bkStock.allBVolRate = stockList.map(_.allBVolRate).sum / stockList.size
    bkStock.agencySVolRate = stockList.map(_.agencySVolRate).sum / stockList.size
    bkStock.allSVolRate = stockList.map(_.allSVolRate).sum / stockList.size
    bkStock.agency1BSVolRate = stockList.map(_.agency1BSVolRate).sum / stockList.size
    bkStock.agency3BSVolRate = stockList.map(_.agency3BSVolRate).sum / stockList.size
    bkStock.agency5BSVolRate = stockList.map(_.agency5BSVolRate).sum / stockList.size
    bkStock.agency10BSVolRate = stockList.map(_.agency10BSVolRate).sum / stockList.size
    bkStock.agency20BSVolRate = stockList.map(_.agency20BSVolRate).sum / stockList.size
    bkStock.aLLVolRate = stockList.map(_.aLLVolRate).sum

    bkStock.all1BSAmt = stockList.map(_.all1BSAmt).sum
    bkStock.all3BSAmt = stockList.map(_.all3BSAmt).sum
    bkStock.all5BSAmt = stockList.map(_.all5BSAmt).sum
    bkStock.all10BSAmt = stockList.map(_.all10BSAmt).sum
    bkStock.minsCnt = stockList.map(_.minsCnt).sum / stockList.size
  }

  /**
    * 计算敏感板块
    *
    * @param bkStock
    * @param stockList
    * @return
    */
  def calcSensitiveBk(bkStock: MinsLine, stockList: ArrayBuffer[MinsLine]): Unit = {
    bkStock.TTime = stockList(0).TTime.substring(0, 4)
    //计算分时逻辑
    bkStock.agencyBVol = stockList.map(_.agencyBVol).sortWith(_ > _).take(15).sum
    bkStock.agencyBAmt = stockList.map(_.agencyBAmt).sortWith(_ > _).take(15).sum
    bkStock.allBVol = stockList.map(_.allBVol).sortWith(_ > _).take(15).sum
    bkStock.allBAmt = stockList.map(_.allBAmt).sortWith(_ > _).take(15).sum
    bkStock.agencySVol = stockList.map(_.agencySVol).sortWith(_ > _).take(15).sum
    bkStock.agencySAmt = stockList.map(_.agencySAmt).sortWith(_ > _).take(15).sum
    bkStock.allSVol = stockList.map(_.allSVol).sortWith(_ > _).take(15).sum
    bkStock.allSAmt = stockList.map(_.allSAmt).sortWith(_ > _).take(15).sum

    bkStock.agencyBSco = stockList.map(_.agencyBSco).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.agencySSco = stockList.map(_.agencySSco).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.priTeamBSco = stockList.map(_.priTeamBSco).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.priTeamSSco = stockList.map(_.priTeamSSco).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.publicBSco = stockList.map(_.publicBSco).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.publicSSco = stockList.map(_.publicSSco).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.allSco = stockList.map(_.allSco).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.avg5AllSco = stockList.map(_.avg5AllSco).sortWith(_ > _).take(15).sum / stockList.take(15).size

    bkStock.agencyBSAmt1D = stockList.map(_.agencyBSAmt1D).sortWith(_ > _).take(15).sum
    bkStock.agencyBSAmt3D = stockList.map(_.agencyBSAmt3D).sortWith(_ > _).take(15).sum
    bkStock.agencyBSAmt5D = stockList.map(_.agencyBSAmt5D).sortWith(_ > _).take(15).sum
    bkStock.agencyBSAmt7D = stockList.map(_.agencyBSAmt7D).sortWith(_ > _).take(15).sum

    //锁定率
    bkStock.agencyBVolRate = stockList.map(_.agencyBVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.allBVolRate = stockList.map(_.allBVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.agencySVolRate = stockList.map(_.agencySVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.allSVolRate = stockList.map(_.allSVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.agency1BSVolRate = stockList.map(_.agency1BSVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.agency3BSVolRate = stockList.map(_.agency3BSVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.agency5BSVolRate = stockList.map(_.agency5BSVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.agency10BSVolRate = stockList.map(_.agency10BSVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.agency20BSVolRate = stockList.map(_.agency20BSVolRate).sortWith(_ > _).take(15).sum / stockList.take(15).size
    bkStock.aLLVolRate = stockList.map(_.aLLVolRate).sortWith(_ > _).take(15).sum

    bkStock.all1BSAmt = stockList.map(_.all1BSAmt).sortWith(_ > _).take(15).sum
    bkStock.all3BSAmt = stockList.map(_.all3BSAmt).sortWith(_ > _).take(15).sum
    bkStock.all5BSAmt = stockList.map(_.all5BSAmt).sortWith(_ > _).take(15).sum
    bkStock.all10BSAmt = stockList.map(_.all10BSAmt).sortWith(_ > _).take(15).sum
    bkStock.minsCnt = stockList.map(_.minsCnt).sum / stockList.size
  }

  /**
    * 输出板块筹码锁定率排名 到kafka
    * @param st
    * @param producer
    */
  def sendSelectStockToKafka(st: MinsLine, producer: KafkaProducer[String, String]) = {
    val (mk, code) = (st.marketType, st.sCode)
    val oldData = oldVolRateTrainMap.getOrElse(mk, new mutable.HashMap[String, ArrayBuffer[Double]]()).get(code)
    oldData match {
      case Some(data) => {
        st.yAgency1BSVolRateTrain = data(1).toInt
      }
      //如果取不到数据历史数据 ??
      case _ =>
    }
    val list = List(st.marketType + "_" + st.sCode, "YAgency1BSVolRateTrain:" + st.yAgency1BSVolRateTrain, "Agency1BSVolRateTrain:" + st.agency1BSVolRateTrain)
    val str = list.mkString("|")
    val record = new ProducerRecord[String, String](Constants.select_stock, st.sCode, str)
    producer.send(record)
  }

  //输出到外部
  def sendOut(lines: ArrayBuffer[MinsLine]) = {
    val producer = Utils.getKafkaProducer(brokers)
    val hbase = Utils.initHbaseTable(Constants.hbase_minsline)
    lines.foreach(
      st => {
        //send to kafka
        sendToKafka(st, producer)
        //send topic to select stock
        sendSelectStockToKafka(st, producer)
        //send to hbase also
        Utils.sendMinsTopValToHbase(hbase, st)
      }
    )
    hbase.close()
  }

  /**
    * 指标数据 写入到 topic.out.MinsLine_Block2
    *
    * @param st
    * @param producer
    */
  def sendToKafka(st: MinsLine, producer: KafkaProducer[String, String]): Unit = {
    val list = List(st.sCode, st.marketType,
      st.agencyBVol, Utils.formatNumber(st.agencyBAmt), st.allBVol, Utils.formatNumber(st.allBAmt), st.agencySVol, Utils.formatNumber(st.agencySAmt), st.allSVol, Utils.formatNumber(st.allSAmt),
      st.agencyBSco, st.agencySSco, st.priTeamBSco, st.priTeamSSco, st.publicBSco,
      st.publicSSco, st.allSco, Utils.formatNumber(st.avg5AllSco),
      Utils.formatNumber(st.agencyBSAmt1D), Utils.formatNumber(st.agencyBSAmt3D),
      Utils.formatNumber(st.agencyBSAmt5D), Utils.formatNumber(st.agencyBSAmt7D), Utils.formatNumber(st.agencyBVolRate), Utils.formatNumber(st.allBVolRate)
      , Utils.formatNumber(st.agencySVolRate), Utils.formatNumber(st.allSVolRate),
      Utils.formatNumber(st.agency1BSVolRate), Utils.formatNumber(st.agency3BSVolRate), Utils.formatNumber(st.agency5BSVolRate), Utils.formatNumber(st.agency10BSVolRate), Utils.formatNumber(st.agency20BSVolRate), Utils.formatNumber(st.aLLVolRate),
      Utils.formatNumber(st.all1BSAmt), Utils.formatNumber(st.all3BSAmt), Utils.formatNumber(st.all5BSAmt), Utils.formatNumber(st.all10BSAmt),
      Utils.formatNumber(st.minsCnt), st.tDate, st.TTime, Utils.getDateTime)
    val str = list.mkString("|")
    val record = new ProducerRecord[String, String](Constants.kafka_out, st.sCode, str)
    producer.send(record)
  }

  /**
    * 收盘逻辑为所有行情数据都已经交易结束 或者 当前时间已经大于下午 4:30
    *
    * @param bkMinsLines
    * @return
    */
  def checkNeedStop(bkMinsLines: ArrayBuffer[MinsLine]) = {
    val notstop = bkMinsLines.filter { p => p.tflag == "0" }
    (bkMinsLines.nonEmpty && notstop.isEmpty) || Utils.hqEnd
  }

  def writeMinsLineToHbase(bkMinsLines: ArrayBuffer[MinsLine]) = {
    val hbaseTable = Utils.getHbaseTable(Constants.hbase_product_fund)
    bkMinsLines.foreach(st => {
      val rowkey = st.sCode.reverse + "_" + st.marketType + "_" + st.tDate
      val row = new Put(Bytes.toBytes(rowkey))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("SCode"), Bytes.toBytes(st.sCode))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("MarketType"), Bytes.toBytes(st.marketType))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencyBVol"), Bytes.toBytes(st.agencyBVol.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencyBAmt"), Bytes.toBytes(Utils.formatNumber(st.agencyBAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllBVol"), Bytes.toBytes(st.allBVol.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllBAmt"), Bytes.toBytes(Utils.formatNumber(st.allBAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencySVol"), Bytes.toBytes(st.agencySVol.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencySAmt"), Bytes.toBytes(Utils.formatNumber(st.agencySAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllSVol"), Bytes.toBytes(st.allSVol.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllSAmt"), Bytes.toBytes(Utils.formatNumber(st.allSAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencyBSco"), Bytes.toBytes(st.agencyBSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencySSco"), Bytes.toBytes(st.agencySSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("PriTeamBSco"), Bytes.toBytes(st.priTeamBSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("PriTeamSSco"), Bytes.toBytes(st.priTeamSSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("PublicBSco"), Bytes.toBytes(st.publicBSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("PublicSSco"), Bytes.toBytes(st.publicSSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllSco"), Bytes.toBytes(st.allSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Avg5AllSco"), Bytes.toBytes(st.avg5AllSco.toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency1BSAmt"), Bytes.toBytes(Utils.formatNumber(st.agencyBSAmt1D).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency3BSAmt"), Bytes.toBytes(Utils.formatNumber(st.agencyBSAmt3D).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency5BSAmt"), Bytes.toBytes(Utils.formatNumber(st.agencyBSAmt5D).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency7BSAmt"), Bytes.toBytes(Utils.formatNumber(st.agencyBSAmt7D).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencyBVolRate"), Bytes.toBytes(Utils.formatNumber(st.agencyBVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllBVolRate"), Bytes.toBytes(Utils.formatNumber(st.allBVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AgencySVolRate"), Bytes.toBytes(Utils.formatNumber(st.agencySVolRate)))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllSVolRate"), Bytes.toBytes(Utils.formatNumber(st.allSVolRate)))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency1BSVolRate"), Bytes.toBytes(Utils.formatNumber(st.agency1BSVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency3BSVolRate"), Bytes.toBytes(Utils.formatNumber(st.agency3BSVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency5BSVolRate"), Bytes.toBytes(Utils.formatNumber(st.agency5BSVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency10BSVolRate"), Bytes.toBytes(Utils.formatNumber(st.agency10BSVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("Agency20BSVolRate"), Bytes.toBytes(Utils.formatNumber(st.agency20BSVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("AllVolRate"), Bytes.toBytes(Utils.formatNumber(st.aLLVolRate).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("All1BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all1BSAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("All3BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all3BSAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("All5BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all5BSAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("All10BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all10BSAmt).toString))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("TDate"), Bytes.toBytes(st.tDate))
      row.addColumn(Bytes.toBytes("cf1"),
        Bytes.toBytes("RTime"), Bytes.toBytes(Utils.getDateTime))
      hbaseTable.put(row)
    })
    hbaseTable.close()
  }

  /**
    * 处理板块数据
    */
  def handler(): Unit = {
    val sleepTime = TimeUnit.MINUTES.toMillis(interval.toLong)
    while (true) {
      try {
        val (bkMinsLines, needStop) = calcMinsLine()
        sendOut(bkMinsLines)
        if (needStop) {
          //写入数据到 product fund
          writeMinsLineToHbase(bkMinsLines)
          println("all stock is end stop process!")
          System.exit(0)
        }
        Thread.sleep(sleepTime)
      } catch {
        case e: Exception => {
          e.printStackTrace()
        }
      }
    }
  }

}
